package com.shujia.flink.source

import org.apache.flink.streaming.api.scala._
import org.apache.flink.api.common.eventtime.WatermarkStrategy
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.kafka.clients.consumer.OffsetResetStrategy

import java.util.Properties

object Demo4KafkaSource {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    /**
     * 构建kafka source
     *
     */

    val properties = new Properties()
    //开启自动提交偏移量
    //消费偏移量会提交到kafk单独的topic中（__consumer_offsets）
    //消费偏移量的保存是以消费者组为单位
    properties.setProperty("enable.auto.commit", "true")

    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      //kafka broker集群列表
      .setBootstrapServers("master:9092")
      //指定其他参数
      .setProperties(properties)
      //执行消费的topic
      .setTopics("test_topic2")
      //指定消费者组，一条数据在一个组内只被消费一次
      .setGroupId("shujia")
      //设置消费数据位置，earliest： 消费所有的数据
      //.setStartingOffsets(OffsetsInitializer.earliest)
      // 从消费组提交的位点开始消费，不指定位点重置策略
      //.setStartingOffsets(OffsetsInitializer.committedOffsets())
      // 从消费组提交的位点开始消费，如果提交位点不存在，使用最早位点
      .setStartingOffsets(OffsetsInitializer.committedOffsets(OffsetResetStrategy.EARLIEST))
      // 从最末尾位点开始消费
      // .setStartingOffsets(OffsetsInitializer.latest());
      //数据的格式，SimpleStringSchema：字符串格式
      .setValueOnlyDeserializer(new SimpleStringSchema())
      .build

    //使用kafka source
    val kafkaDS: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")


    kafkaDS.print()

    env.execute()

  }

}
